package com.smsc.headend.task.engine.utils;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import com.smsc.headend.common.constant.Constant;
import com.smsc.headend.module.mdm.dds.dto.MdmAnonyTerminalDto;
import com.smsc.headend.task.engine.manager.DdsFilterMdmDataManager;
import com.smsc.headend.task.engine.service.asset.AssetService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @author wxh
 * @date 2021/4/13 19:25
 */
@Component
@Slf4j
public class DdsAnonymousTerminalStatusGeneUtil {
    ScheduledExecutorService persistentService;

    BlockingQueue<MdmAnonyTerminalDto> ddsQueue = new LinkedBlockingQueue();

    @Autowired
    DdsFilterMdmDataManager ddsFilterMdmDataManager;
    @Autowired
    AssetService assetService;

    @PostConstruct
    public void initSchedule() {
        persistentService = Executors.newSingleThreadScheduledExecutor();
        persistentService.scheduleAtFixedRate(() -> {
            Long start = SystemClock.now();
            Integer size = 0;
            try {
                List<MdmAnonyTerminalDto> readDatas = new ArrayList<>();
                while (true) {
                    MdmAnonyTerminalDto readData = ddsQueue.poll(5, TimeUnit.SECONDS);
                    if (readData == null) {
                        break;
                    }
                    readDatas.add(readData);
                    String ddsFileBatchSize = "1000";
                    Integer batchThreshold = 1000;
                    try {
                        batchThreshold = Convert.toInt(ddsFileBatchSize);
                    } catch (Exception e) {
                        log.error("ddsFileBatchSize system error {}, using defalut {}", ddsFileBatchSize, batchThreshold);
                    }
                    if (readDatas.size() == batchThreshold) {
                        break;
                    }
                }
                if (CollectionUtils.isEmpty(readDatas)) {
                    return;
                }
                size = readDatas.size();
                //是否上传电力MDM
                String mdmFlag = assetService.getSystemProperty("SYNC_MDM_SWITCH");
                if (Constant.TRUE.equalsIgnoreCase(mdmFlag)) {
                    Map<Integer, List<String>> mdmDataMap = ddsFilterMdmDataManager.transformDdsAnonyDeviceStatusForMdm(readDatas);
                    if (CollectionUtil.isNotEmpty(mdmDataMap)) {
                        ddsFilterMdmDataManager.generateTxtFilesAndUploadTheSelectedType(mdmDataMap);
                    }
                }
                log.info("save dds anonymous terminal status cost: {}ms, dataSize={} ,left={} ", SystemClock.now() - start, size, ddsQueue.size());
            } catch (Exception e) {
                log.error("dds save failed", e);
            }
        }, 0, 200, TimeUnit.MILLISECONDS);
    }

    public void insertRecord(MdmAnonyTerminalDto record) {
        try {
            if (!ddsQueue.offer(record)) {
                log.error("no data");
            }
        }
        catch (Exception e) {
            log.error("insert error", record, e);
        }
    }
}
